package pku;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.ArrayList;
import java.util.HashMap;

/**
 * Created by yangxiao on 2017/11/14. 这是一个消息队列的内存实现
 */
public class MessageStore {
	static final MessageStore store = new MessageStore();
	// 消息存储
	HashMap<String, ArrayList<ByteMessage>> msgs = new HashMap<String, ArrayList<ByteMessage>>();
	HashMap<Long, HashMap<Long, Thread>> threadMap = new HashMap<Long, HashMap<Long, Thread>>();
	HashMap<String, Integer> topicCount = new HashMap<String, Integer>();
	HashMap<String, HashMap<String, Integer>> readTopicCount = new HashMap<String, HashMap<String, Integer>>();
	// 记录消息大小
	long totalbyte = 0l;
	static final String filePath = "./data/";
//	 static final String filePath = "C:\\Users\\Administrator\\git\\javamq-hb\\data\\";

	// 加锁保证线程安全
	public synchronized void push(ByteMessage msg, String topic) {
		if (msg == null) {
			return;
		}
		if (!msgs.containsKey(topic)) {
			msgs.put(topic, new ArrayList<>());
		}
		// 加入消息
		msgs.get(topic).add(msg);
		totalbyte = totalbyte + msg.getBody().length;
		if (totalbyte / 1024 / 1024 > 15) {
			totalbyte = 0l;
			MessageStore.store.writeObjectToFile(msgs, "1");
		}

	}

	public synchronized void writeObjectToFile(Object obj, String typeFlag) {
		HashMap<String, ArrayList<ByteMessage>> temp = MessageStore.store.msgs;
		MessageStore.store.msgs = new HashMap<String, ArrayList<ByteMessage>>();

		HashMap<String, String> tempCount = new HashMap<String, String>();
		for (String topic : temp.keySet()) {
			if (MessageStore.store.topicCount.containsKey(topic)) {
				MessageStore.store.topicCount.put(topic, MessageStore.store.topicCount.get(topic) + 1);
			} else {
				MessageStore.store.topicCount.put(topic, 0);
			}
			tempCount.put(topic, topic + "-" + MessageStore.store.topicCount.get(topic));
		}
		Thread t = new WriteObjectThread(tempCount, temp, Thread.currentThread().getId(), typeFlag);
		t.start();

		if (this.threadMap.containsKey(Thread.currentThread().getId())) {
			this.threadMap.get(Thread.currentThread().getId()).put(t.getId(), t);
		} else {
			HashMap<Long, Thread> threadMapTemp = new HashMap<Long, Thread>();
			threadMapTemp.put(t.getId(), t);
			this.threadMap.put(Thread.currentThread().getId(), threadMapTemp);
		}
		while (true) {
			if ((this.threadMap.get(Thread.currentThread().getId()).keySet()).size() > 2) {
				try {
					Thread.sleep(300);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			} else {
				break;
			}
		}
	}

	@SuppressWarnings({ "unchecked", "unused" })
	public synchronized ArrayList<ByteMessage> readObjectFromFile(String topic) {
		ArrayList<ByteMessage> temp = null;
		if (this.topicCount.isEmpty()) {
			this.topicCount = (HashMap<String, Integer>) this.readObject("topicindex");
		}
		String strKey = Thread.currentThread().getId() + topic;

		if (this.readTopicCount.containsKey(strKey)) {
			if (this.readTopicCount.get(strKey).get("currentCount") < this.readTopicCount.get(strKey)
					.get("totalCount")) {
				this.readTopicCount.get(strKey).put("currentCount",
						this.readTopicCount.get(strKey).get("currentCount") + 1);
				return this.msgs.get(strKey);
			} else {
				int index = this.readTopicCount.get(strKey).get("index");
				if (index >= this.topicCount.get(topic)) {// 当前topic已读取完毕
					this.msgs.remove(strKey);
					this.readTopicCount.remove(strKey);
					return null;
				}
				HashMap<String, ArrayList<ByteMessage>> msgTemp = (HashMap<String, ArrayList<ByteMessage>>) this
						.readObject(topic + "-" + (index + 1));
				if (null != msgTemp) {
					this.msgs.put(strKey, msgTemp.get(topic));
					this.readTopicCount.get(strKey).put("index", index + 1);
					this.readTopicCount.get(strKey).put("currentCount", 1);
					this.readTopicCount.get(strKey).put("totalCount", msgTemp.get(topic).size());
				} else {
					return null;
				}
			}
		} else {
			HashMap<String, ArrayList<ByteMessage>> msgTemp = (HashMap<String, ArrayList<ByteMessage>>) this
					.readObject(topic + "-0");
			if (null != msgTemp) {
				HashMap<String, Integer> detailMap = new HashMap<String, Integer>();
				this.readTopicCount.put(strKey, detailMap);
				this.msgs.put(strKey, msgTemp.get(topic));
				detailMap.put("index", 0);
				detailMap.put("currentCount", 1);
				detailMap.put("totalCount", msgTemp.get(topic).size());
			} else {
				return null;
			}
		}

		temp = this.msgs.get(strKey);
		return temp;
	}

	public Object readObject(String fileName) {
		FileInputStream fis = null;
		ObjectInputStream ois = null;
		BufferedInputStream bis = null;
		File file = null;
		Object object = null;
		try {
			file = new File(MessageStore.filePath + fileName + ".hb");
			if (!file.exists()) {
				return null;
			}
			fis = new FileInputStream(file);
			bis = new BufferedInputStream(fis);
			ois = new ObjectInputStream(bis);
			object = ois.readObject();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		} finally {
			try {
				if (null != ois) {
					ois.close();
				}
				if (null != bis) {
					bis.close();
				}
				if (null != fis) {
					fis.close();
				}
				
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return object;
	}

}
